package com.zhang.third.day10;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @title: 实时热门商品--FlinkSQL实现
 * @author: zhang
 * @date: 2022/4/15 09:38
 */
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 创建数据源表
        tableEnv.executeSql("" +
                "create table item ( " +
                "userId string," +
                "pId string," +
                "cId string," +
                "type string," +
                "ts bigint," +
                "rt as TO_TIMESTAMP(FROM_UNIXTIME(ts))," +
                "WATERMARK FOR rt as rt - INTERVAL '0' SECOND ) " +
                "with (" +
                "'connector' = 'filesystem', " +
                "'path' = 'file:///Users/apple/IdeaProjects/flink_1.13/src/main/resources/UserBehavior.csv'," +
                "'format' = 'csv' )");


        // 计算每个商品在所属窗口的pv
        //TODO 分组开窗
        Table itemCountWindow = tableEnv.sqlQuery("" +
                "select " +
                " pId, " +
                " count(pId) as cnt, " +
                " window_end ent " +
                " from TABLE(" +
                " HOP( TABLE item," +
                "DESCRIPTOR(rt)," +
                "INTERVAL '5' MINUTES,INTERVAL '1' HOURS))" +
                "GROUP BY pId,window_start,window_end");

        //tableEnv.toChangelogStream(itemCountWindow).print();
        // TODO over()开窗排序
        Table rkTable = tableEnv.sqlQuery(
                "select " +
                        " pId," +
                        " cnt," +
                        " ent windowEnd," +
                        " row_number() over(partition by ent order by cnt desc) as rk" +
                        " from " + itemCountWindow
        );

        // todo  求TopN
        Table result = tableEnv.sqlQuery(
                "select" +
                        " * " +
                        "from " +
                        "" + rkTable + " " +
                        "where rk <= 3");
        //todo 打印表信息
        tableEnv.toChangelogStream(result).print();

        env.execute();
    }
}
